package com.etlmaster.datax.plugins.reader.localfilereader;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;

import org.apache.commons.vfs2.FileObject;

import com.dp.nebula.wormhole.common.AbstractPlugin;
import com.dp.nebula.wormhole.common.JobStatus;
import com.dp.nebula.wormhole.common.interfaces.ILine;
import com.dp.nebula.wormhole.common.interfaces.ILineSender;
import com.dp.nebula.wormhole.common.interfaces.IReader;
import com.etlmaster.executor.utils.LogWriter;

public class LocalFileReader extends AbstractPlugin implements IReader{
	
	static final int PLUGIN_NO = 1;
	
	static final int ERROR_CODE_ADD = JobStatus.PLUGIN_BASE*PLUGIN_NO;
	
	private String filePath;
	
	private String dirPath;

	private String fileName;

	private char colDelimiter = '\t';

	private String encoding = "UTF-8";
	
	private String colNum = "0";
	
	private String nullString = "";
	
	
	
	private FileObject fo = null;
	
	private InputStream in = null;
	
	private BufferedReader br = null;
	
	private int bufferSize = 4 * 1024;
	public static final int LINE_MAX_FIELD = 1024;
	private int[] colList = new int[LINE_MAX_FIELD];
	
	@Override
	public void init() {
		this.filePath = getParam().getValue(ParamKey.filePath,"");
		this.dirPath = getParam().getValue(ParamKey.dirPath,"");
		this.nullString = getParam().getValue(ParamKey.nullString,"");
		this.fileName = getParam().getValue(ParamKey.fileName,"");
		this.colDelimiter = getParam().getCharValue(ParamKey.colDelimiter,'\t');
		this.encoding = getParam().getValue(ParamKey.encoding,"UTF-8");
		this.colNum = getParam().getValue(ParamKey.colNum,"");
	}

	@Override
	public void connection() {
		
		try {
			 File f = new File(filePath);   
		    in = new FileInputStream(f);   
			br = new BufferedReader(
			        new InputStreamReader(in, encoding), bufferSize);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void read(ILineSender lineSender){
		String lineString = null;
		try {

			
			while ((lineString = br.readLine()) != null){
				ILine line = lineSender.createNewLine();
				int i = 0;
				int begin = 0;
				int length = lineString.length();

					for (i = 0; i < length; ++i) {
						if (lineString.charAt(i) == colDelimiter) {
							line.addField(replaceNullString(lineString.substring(begin, i)));
							begin = i + 1;
						}
					}
					line.addField(replaceNullString(lineString.substring(begin, i)));
					
				boolean flag = lineSender.send(line);
				if(getMonitor()!=null) {
					if (flag){
						getMonitor().increaseSuccessLines();
					}else{
						getMonitor().increaseFailedLines();
					}
				}
			}
		} catch (IOException e) {
			LogWriter.addLog("ERROR",e.getMessage());
		}
		lineSender.flush();
	}
	
	

	
	private String replaceNullString(String string) {
		if (nullString != null && nullString.equals(string)) {
			return null;
		}
		return string;
	}
	
	@Override
	public void finish(){
		closeAll();
	}
	
	public void chkFileRows(){
		
	}
	
	public void closeAll(){
		try {
			if (br != null) {
				br.close();
			}
			if (in != null) {
				in.close();
			}
			if(fo!=null){
				fo.close();
			}
		} catch (IOException e) {
			LogWriter.addLog("ERROR",e.getMessage());
		}
	}
}
